#ifndef __M_BROKER_H__
#define __M_BROKER_H__
#include "../mqthird/include/muduo/proto/codec.h"
#include "../mqthird/include/muduo/proto/dispatcher.h"

#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include"mq_connection.hpp"
#include"mq_virtualhost.hpp"
#include"mq_consumer.hpp"
#include"../mqcommon/mq_threadpool.hpp"
#include"../mqcommon/mq_logger.hpp"
#include"../mqcommon/mq_proto.pb.h"
#include"mq_channel.hpp"
#include <iostream>

#define DBFILE "./meta.db"
class Server
{

public:
    Server(int port,std::string basedir) : _server(&_loop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort),
                       _dispatcher(std::bind(&Server::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
                       _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
                       _valtualhost(std::make_shared<bitmq::VirtualHost>(basedir,basedir+DBFILE)),
                       _consumer_manager(std::make_shared<bitmq::ConsumerManager>()),
                       _connection_manager(std::make_shared<bitmq::ConnectionManager>()),
                       _threadpool(std::make_shared<threadpool>())
    {
        //把所有的消费者队列都初始化，防止创建消费者失败
        bitmq::MsgQueueMap msgq=_valtualhost->getQueues();
        for(auto q:msgq)
        {
            _consumer_manager->initQueueConsumer(q.first);
        }

        _dispatcher.registerMessageCallback<bitmq::openChannelRequest>(
            std::bind(&Server::onopenChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::closeChannelRequest>(
            std::bind(&Server::oncloseChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

         _dispatcher.registerMessageCallback<bitmq::declareExchangeRequest>(
          std::bind(&Server::ondeclareExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); 
         _dispatcher.registerMessageCallback<bitmq::deleteExchangeRequest>(
                    std::bind(&Server::ondeleteExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

         _dispatcher.registerMessageCallback<bitmq::declareQueueRequest>(
            std::bind(&Server::ondeclareQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<bitmq::deleteQueueRequest>(
             std::bind(&Server::ondeleteQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

          _dispatcher.registerMessageCallback<bitmq::queueBindingRequest>(
             std::bind(&Server::onqueueBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<bitmq::queueUnBindingRequest>(
             std::bind(&Server::onqueueUnBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

         _dispatcher.registerMessageCallback<bitmq::basicPublishRequest>(
                 std::bind(&Server::onbasicPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
          _dispatcher.registerMessageCallback<bitmq::basicAckRequest>(
                 std::bind(&Server::onbasicAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

          _dispatcher.registerMessageCallback<bitmq::basicConsumerRequest>(
                  std::bind(&Server::onbasicConsumer, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
          _dispatcher.registerMessageCallback<bitmq::basicConcelRequest>(
                  std::bind(&Server::onbasicCancel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                                   std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
    }
    void Start()
    {
        _server.start();
        _loop.loop();
    }
    //信道的打开
    void onopenChannel(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::openChannelRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("信道打开失败，没找到连接");
                conn->shutdown();
                return;
            }
            return connection_ptr->openChannel(message);
        }
    //信道的关闭
    void oncloseChannel(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::closeChannelRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("信道关闭失败，没找到连接");
                conn->shutdown();
                return;
            }
            return connection_ptr->closeChannel(message);
        }
    //交换机的定义
    void ondeclareExchange(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::declareExchangeRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("交换机定义失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("交换机定义失败,没找到信道");
                return;
            }
            channel->declareExchange(message);
        }
  
    //交换机的删除
    void ondeleteExchange(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::deleteExchangeRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("交换机删除失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("交换机删除失败,没找到信道");
                return;
            }
            channel->deleteExchange(message);
        }
    //队列的定义
    void ondeclareQueue(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::declareQueueRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("队列定义失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("队列定义失败,没找到信道");
                return;
            }
            channel->declareQueue(message);
        }
    //队列的删除
    void ondeleteQueue(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::deleteQueueRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("队列删除失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("队列删除失败,没找到信道");
                return;
            }
            channel->deleteQueue(message);
        }
    //队列的绑定
    void onqueueBind(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::queueBindingRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("队列绑定失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("队列绑定失败,没找到信道");
                return;
            }
            channel->bind(message);
        }
    //队列的解绑
    void onqueueUnBind(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::queueUnBindingRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("队列解绑失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("队列解绑失败,没找到信道");
                return;
            }
            channel->unbind(message);
        }
    //消息的发布
    void onbasicPublish(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::basicPublishRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("消息发布失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("消息发布失败,没找到信道");
                return;
            }
            channel->basicPublish(message);
        }
    //消息的确认
    void onbasicAck(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::basicAckRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("消息确认失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("消息确认失败,没找到信道");
                return;
            }
            channel->basicAck(message);
        }
    //消息的订阅
    void onbasicConsumer(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::basicConsumerRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("消息订阅失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("消息订阅失败,没找到信道");
                return;
            }
            channel->basicConsumer(message);
        }
    //取消订阅
    void onbasicCancel(const muduo::net::TcpConnectionPtr &conn,
        const bitmq::basicConcelRequestPtr &message,
        muduo::Timestamp)
        {
            bitmq::Connection::ptr connection_ptr=_connection_manager->getConnection(conn);
            if(connection_ptr.get()==nullptr)
            {
                DLOG("取消订阅失败，没找到连接");
                conn->shutdown();
                return;
            }

            bitmq::Channel::ptr channel=connection_ptr->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("取消订阅失败,没找到信道");
                return;
            }
            channel->basicConcel(message);
        }


    void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,
                          const MessagePtr &message, muduo::Timestamp)
    {
        LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
        conn->shutdown();
    }


    void onConnection(const muduo::net::TcpConnectionPtr &conn)
    {
        if (conn->connected())
        {
            DLOG("新连接建立成功！");
            _connection_manager->newConnection(conn,_codec,_valtualhost,_threadpool,_consumer_manager);
        }
        else
        {
            DLOG("连接即将关闭！");
          _connection_manager->deleteConnection(conn);
        }
    }

private:
    muduo::net::EventLoop _loop;
    muduo::net::TcpServer _server;  // 服务器
    ProtobufDispatcher _dispatcher; // 请求分发器对象--要向其注册处理函数
    bitmq::ProtobufCodecPtr _codec;           // protobuf协议处理器---针对收到的数据进行对应的协议处理

    bitmq::VirtualHost::ptr _valtualhost;  //虚拟机管理句柄
    bitmq::ConnectionManager::ptr _connection_manager;//muduo库中连接的管理句柄
    bitmq::ConsumerManager::ptr _consumer_manager;//消费者管理句柄
    threadpool::ptr _threadpool;   //线程池管理句柄
};
#endif